package com.niit.covid.MeiTuanStaterjoin;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class JoinMapper extends Mapper<LongWritable, Text,Text,Text> {
    Text outKey = new Text();
    Text outValue = new Text();
    StringBuilder sb = new StringBuilder();
    String filename = null;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //获取当前处理的切片所属的文字名字
        FileSplit inputSplit = (FileSplit) context.getInputSplit();
        filename = inputSplit.getPath().getName();
        System.out.println("当前正在处理的文件十："+filename);
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //设置字符串长度，用于清空数据
        sb.setLength(0);

        //切割处理输入数据
        String[] fields = value.toString().split("\\|");
        //判断处理的是哪个文件
        if (filename.contains("itheima_goods.txt")){
            //处理的是商品数据
            outKey.set(fields[0]);
            StringBuilder append = sb.append(fields[1]).append("\t").append(fields[2]);
            outValue.set(sb.insert(0,"goods#").toString());
            System.out.println(outKey + "---->" + outValue);
            context.write(outKey,outValue);
        }
        else {
            //处理的是订单数据
            outKey.set(fields[1]);
            StringBuilder append = sb.append(fields[0]).append("\t").append(fields[2]);
            outValue.set(sb.insert(0,"order#").toString());
            System.out.println(outKey + "---->" + outValue);
        }
    }
}
